package com.ziroom.crm.mq.rabbit.spring.core;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.ziroom.crm.mq.core.reconsume.ReConsumeInvoker;
import com.ziroom.crm.mq.core.reconsume.ReConsumeMethodHolder;
import com.ziroom.crm.mq.core.retry.RetryContext;
import com.ziroom.crm.mq.core.support.ProcessorInvokerDelegate;
import com.ziroom.crm.mq.core.support.SimpleProcessorInvokerDelegate;
import com.ziroom.crm.mq.rabbit.spring.data.MqRabbitConst;
import com.ziroom.crm.mq.rabbit.spring.data.RabbitMessage;
import com.ziroom.crm.mq.rabbit.spring.resolve.RabbitResolve;
import com.ziroom.crm.mq.rabbit.spring.wrapper.RabbitConsumeQueueWrapper;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;

import java.util.List;
import java.util.Objects;
import java.util.function.Consumer;

/**
 * 消息容器
 *
 * @author WeiShaoying
 * @version 1.0
 * @date Created in 2020年09月06日 14:59
 * @since 1.0
 */
@Slf4j
public class ReliableMessageListenerContainer extends SimpleMessageListenerContainer {

    private final ProcessorInvokerDelegate facade = new SimpleProcessorInvokerDelegate();

    @Getter
    @Setter
    public Consumer<List<? extends com.ziroom.crm.mq.core.data.Message>> callback;

    @Getter
    @Setter
    public Consumer<Integer> countCallback;

    @Override
    public void initialize() {
        MessageListener messageListener = (MessageListener) super.getMessageListener();
        for (String queueName : super.getQueueNames()) {
            log.info("初始化队列：{}", queueName);
            RabbitConsumeQueueWrapper queueWrapper = RabbitResolve.buildConsumeQueueWrapper(queueName);
            // 标记处理方法
            ReConsumeMethodHolder.addObject(queueWrapper, messageListener);
        }
        if (Objects.nonNull(callback)) {
            ReConsumeInvoker.setCallback(callback);
        }
        if (Objects.nonNull(countCallback)) {
            ReConsumeInvoker.setCountCallback(countCallback);
            //log.debug("setCountCallback success: {}", countCallback);
        }
        super.initialize();
    }

    @Override
    protected void doInvokeListener(MessageListener listener, Message message) {
        RabbitMessage rabbitMessage = RabbitResolve.convert(message);
        rabbitMessage.setQueueName(getQueueName(message));

        RabbitConsumeQueueWrapper consumeQueueWrapper = RabbitResolve.buildConsumeQueueWrapper(rabbitMessage);
        // 设置回调
        consumeQueueWrapper.setRetryCallback((retryContext) -> {
            // 消息中设置内存重试次数
            message.getMessageProperties().setHeader(MqRabbitConst.MQ_CONSUME_RETRY_COUNT, retryContext.getRetryCount());
            super.doInvokeListener(listener, message);
        });
        facade.preConsumeInvoke(consumeQueueWrapper, rabbitMessage);
        try {
            super.doInvokeListener(listener, message);
        } catch (Exception e) {
            log.error("[crm-mq]消费失败，messageId:{}", rabbitMessage.getId(), e);
            rabbitMessage.setException(e);
        } finally {
            facade.postConsumeInvoke(consumeQueueWrapper, rabbitMessage);
        }
    }

    @Override
    protected void doInvokeListener(ChannelAwareMessageListener listener, Channel channel, Message message) {
        RabbitMessage rabbitMessage = RabbitResolve.convert(message);
        rabbitMessage.setQueueName(getQueueName(message));

        RabbitConsumeQueueWrapper rabbitQueueWrapper = RabbitResolve.buildConsumeQueueWrapper(rabbitMessage);
        facade.preConsumeInvoke(rabbitQueueWrapper, rabbitMessage);

        try {
            // Channel通道
            super.doInvokeListener(listener, channel, message);
        } catch (Exception e) {
            log.error("[crm-mq]channel消费失败，messageId:{}", rabbitMessage.getId(), e);
            rabbitMessage.setException(e);
        } finally {
            Consumer<RetryContext> consumer = (RetryContext retryContext) -> {
                //消息中设置内存重试次数
                message.getMessageProperties().setHeader(MqRabbitConst.MQ_CONSUME_RETRY_COUNT, retryContext.getRetryCount());
                super.doInvokeListener(listener, channel, message);
            };
            rabbitQueueWrapper.setRetryCallback(consumer);
            facade.postConsumeInvoke(rabbitQueueWrapper, rabbitMessage);
        }
    }

    /**
     * 通过Spring获得队列名称
     *
     * @param message 消息
     * @return 队列名
     */
    private String getQueueName(Message message) {
        return message.getMessageProperties().getConsumerQueue();
    }

    @Override
    public void stop() {
        log.info("Stop Queue Name: {}", JSON.toJSONString(super.getQueueNames()));
        //注销监听方法 避免内存泄露
        for (String queueName : super.getQueueNames()) {
            RabbitConsumeQueueWrapper consumeQueueWrapper = RabbitResolve.buildConsumeQueueWrapper(queueName);
            ReConsumeMethodHolder.removeObject(consumeQueueWrapper);
        }
        super.stop();
    }

}
